package com.hrt.iceberg

import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * Spark 向Iceberg中写入数据，包含：
  * 1) merge into
  * 2) insert overwrite
  * 3) Delete from
  * 4) Update
  */
object SparkIcebergWrites {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder().master("local").appName("SparkOperateIceberg")
      //指定hadoop catalog，catalog名称为hadoop_prod
      .config("spark.sql.catalog.hadoop_prod", "org.apache.iceberg.spark.SparkCatalog")
      .config("spark.sql.catalog.hadoop_prod.type", "hadoop")
      .config("spark.sql.catalog.hadoop_prod.warehouse", "hdfs://mycluster/sparkoperateiceberg")
      .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
      .getOrCreate()

    /**
      * Merge into
      */
    //创建一张表 a ，并插入数据
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.a (id int,name string,age int) using iceberg
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.a values (1,"zs",18),(2,"ls",19),(3,"ww",20)
//      """.stripMargin)

    //创建另外一张表b ,并插入数据
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.b (id int,name string,age int,tp string) using iceberg
//      """.stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.b values (1,"zs",30,"delete"),(2,"李四",31,"update"),(4,"王五",32,"add")
//      """.stripMargin)

    //将表b 中与表a中相同id的数据更新到表a,表a中没有表b中有的id对应数据写入增加到表a
//    spark.sql(
//      """
//        |MERGE INTO hadoop_prod.default.a  t1
//        |USING (SELECT id,name ,age,tp FROM hadoop_prod.default.b) t2
//        |ON t1.id = t2.id
//        |WHEN MATCHED AND t2.tp = 'delete' THEN DELETE
//        |WHEN MATCHED AND t2.tp = 'update' THEN UPDATE SET t1.name = t2.name,t1.age = t2.age
//        |WHEN NOT MATCHED THEN INSERT (id,name,age) VALUES (t2.id,t2.name,t2.age)
//      """.stripMargin)
//
//    spark.sql("""select * from hadoop_prod.default.a """).show()


    /**
      * insert overwrite
     */
    //创建 test1 分区表，并插入数据
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.test1 (id int,name string,loc string)
//        |using iceberg
//        |partitioned by (loc)
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.test1 values (1,"zs","beijing"),(2,"ls","shanghai")
//      """.stripMargin)


    //创建 test2 普通表，并插入数据
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.test2 (id int,name string,loc string)
//        |using iceberg
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.test2 values (10,"x1","shandong"),(11,"x2","hunan")
//      """.stripMargin)


    //创建 test3 普通表，并插入数据
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.test3 (id int,name string,loc string)
//        |using iceberg
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.test3 values (3,"ww","beijing"),(4,"ml","shanghai"),(5,"tq","guangzhou")
//      """.stripMargin)

    //使用insert overwrite 读取test3 表中的数据覆盖到test2 普通表中
//    spark.sql(
//      """
//        |insert overwrite hadoop_prod.default.test2
//        |select id,name,loc from  hadoop_prod.default.test3
//      """.stripMargin)

    //查询 test2 表中的数据
//    spark.sql(
//      """
//        |select * from hadoop_prod.default.test2
//      """.stripMargin).show()

    // 使用insert overwrite 读取test3表数据 动态分区方式覆盖到表 test1
//    spark.sql(
//      """
//        |insert overwrite hadoop_prod.default.test1
//        |select id,name,loc from  hadoop_prod.default.test3
//      """.stripMargin)

    //查询 test1 表数据
//    spark.sql(
//      """
//        |select * from hadoop_prod.default.test1
//      """.stripMargin).show()


    // 使用insert overwrite 读取test3表数据 静态分区方式覆盖到表 test1，表中其他分区数据不受影响，只会覆盖指定的静态分区数据
    //删除表test1,重新创建表test1 分区表，并插入数据
//    spark.sql(
//      """
//        |drop table hadoop_prod.default.test1
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |create table  hadoop_prod.default.test1 (id int,name string,loc string)
//        |using iceberg
//        |partitioned by (loc)
//      """.stripMargin)
//
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.test1 values (1,"zs","beijing"),(2,"ls","shanghai")
//      """.stripMargin)
//
//    spark.sql("select * from hadoop_prod.default.test1").show()


    //注意：指定静态分区"jiangsu",静态分区下，就不要在查询 “loc" 列了，否则重复
//    spark.sql(
//      """
//        |insert overwrite hadoop_prod.default.test1
//        |partition (loc = "jiangsu")
//        |select id,name from  hadoop_prod.default.test3
//      """.stripMargin)

    //查询 test1 表数据
//    spark.sql(
//      """
//        |select * from hadoop_prod.default.test1
//      """.stripMargin).show()

    /**
      * Delete from
      */
    //创建表 delete_tbl ,并加载数据
//    spark.sql(
//      """
//        |create table hadoop_prod.default.delete_tbl (id int,name string,age int) using iceberg
//        |""".stripMargin)

//    spark.sql(
//      """
//        |insert into hadoop_prod.default.delete_tbl values (1,"zs",18),(2,"ls",19),(3,"ww",20),(4,"ml",21),(5,"tq",22),(6,"gb",23)
//      """.stripMargin)


    //根据条件范围删除表 delete_tbl 中的数据
//    spark.sql(
//      """
//        |delete from hadoop_prod.default.delete_tbl where id >3 and id <6
//      """.stripMargin)
//
//    spark.sql("select * from hadoop_prod.default.delete_tbl").show()

    //根据条件删除表 delete_tbl 中的一条数据
//    spark.sql(
//      """
//        |delete from hadoop_prod.default.delete_tbl where id = 2
//      """.stripMargin)
//
//    spark.sql("select * from hadoop_prod.default.delete_tbl").show()

    /**
      * update
      */
    //创建表 delete_tbl ,并加载数据
//    spark.sql(
//      """
//        |create table hadoop_prod.default.update_tbl (id int,name string,age int) using iceberg
//        |""".stripMargin)
//    spark.sql(
//      """
//        |insert into hadoop_prod.default.update_tbl values (1,"zs",18),(2,"ls",19),(3,"ww",20),(4,"ml",21),(5,"tq",22),(6,"gb",23)
//      """.stripMargin)

    //更新 delete_tbl 表
//    spark.sql(
//      """
//        |update hadoop_prod.default.update_tbl set name = 'zhangsan' , age = 30
//        |where id <=3
//      """.stripMargin)
//    spark.sql(
//      """
//        |select * from hadoop_prod.default.update_tbl
//      """.stripMargin).show()

    /**
      * Spark DataFrame Api 操作Iceberg
      */
    //1.准备数据，使用DataFrame Api 写入Iceberg表及分区表
//    val nameJsonList = List[String](
//      "{\"id\":1,\"name\":\"zs\",\"age\":18,\"loc\":\"beijing\"}",
//      "{\"id\":2,\"name\":\"ls\",\"age\":19,\"loc\":\"shanghai\"}",
//      "{\"id\":3,\"name\":\"ww\",\"age\":20,\"loc\":\"beijing\"}",
//      "{\"id\":4,\"name\":\"ml\",\"age\":21,\"loc\":\"shanghai\"}")

//    import spark.implicits._
//    val df: DataFrame = spark.read.json(nameJsonList.toDS)

    //创建普通表df_tbl1,并将数据写入到Iceberg表，其中DF中的列就是Iceberg表中的列
//    df.writeTo("hadoop_prod.default.df_tbl1").create()

    //查询表 hadoop_prod.default.df_tbl1 中的数据，并查看数据存储结构
//    spark.read.table("hadoop_prod.default.df_tbl1").show()



    //创建分区表df_tbl2,并将数据写入到Iceberg表，其中DF中的列就是Iceberg表中的列
//    df.sortWithinPartitions($"loc")//写入分区表，必须按照分区列进行排序
//      .writeTo("hadoop_prod.default.df_tbl2")
//      .partitionedBy($"loc")//这里可以指定多个列为联合分区
//      .create()
    //查询分区表 hadoop_prod.default.df_tbl2 中的数据，并查看数据存储结构
//    spark.read.table("hadoop_prod.default.df_tbl2").show()

  }
}
